package edu.wisc.hevea.broker;

import java.io.IOException;

import org.apache.zookeeper.KeeperException;
import org.apache.zookeeper.ZooKeeper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import edu.wisc.hevea.constants.ApplicationConstants;
import edu.wisc.hevea.zookeeper.election.LeaderElectionAware;
import edu.wisc.hevea.zookeeper.election.LeaderElectionSupport;
import edu.wisc.hevea.zookeeper.election.LeaderElectionSupport.EventType;
import edu.wisc.hevea.zookeeper.util.ZooKeeperUtil;

public class LeaderElectionThread implements Runnable, LeaderElectionAware {

	private static final Logger logger = LoggerFactory
			.getLogger(LeaderElectionThread.class);

	public static final String LEADER_NODE = "LEADER";
	private String leaderPath;

	private MessageBroker messageBroker;
	private String rootNode;

	private ZooKeeper currentZooKeeper;

	public LeaderElectionThread(MessageBroker messageBroker) {
		this.messageBroker = messageBroker;
		this.rootNode = messageBroker.getApplicationPolicy().getApplicationId();
		this.setLeaderPath("/" + rootNode + "/" + LEADER_NODE);
	}

	@Override
	public void run() {

		int sessionTimeout = 3000;
		String myHostName = messageBroker.getMyHostName();
		String zooKeeperhostname = ZooKeeperUtil
				.getPreferredZookeeperHostName(myHostName);

		logger.info("ZooKeeperHostname : " + zooKeeperhostname);

		LeaderElectionSupport leaderElectionSupport = new LeaderElectionSupport();
		try {

			currentZooKeeper = ZooKeeperUtil.getZooKeeperInstance(
					zooKeeperhostname, sessionTimeout, leaderElectionSupport);

			String appRoot = "/" + rootNode;
			String electionPath = appRoot + "/"
					+ ApplicationConstants.ELECTION_PREFIX;
			if (!doesAppPathExist()) {
				ZooKeeperUtil.createPath(currentZooKeeper, appRoot, rootNode);
			}
			if (!doesElectionPathExist(electionPath)) {
				ZooKeeperUtil.createPath(currentZooKeeper, electionPath,
						"ELECTION");
			}

			leaderElectionSupport.setZooKeeper(currentZooKeeper);
			leaderElectionSupport.setHostName(myHostName);
			leaderElectionSupport.setRootNodeName(electionPath);
			leaderElectionSupport.addListener(this);
			leaderElectionSupport.start();

			// If we don't get response by this time
			// we have to change preferred server and start over
			// Thread.sleep(30 * 1000);

		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();

		} catch (KeeperException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		} catch (InterruptedException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	private boolean doesElectionPathExist(String electionPath)
			throws KeeperException, InterruptedException {
		return ZooKeeperUtil.doesExistPath(currentZooKeeper, electionPath);
	}

	private boolean doesAppPathExist() throws KeeperException,
			InterruptedException {
		return ZooKeeperUtil.doesExistPath(currentZooKeeper,
				getApplicationPath());
	}

	private String getApplicationPath() {
		return String.format("%s%s", "/", messageBroker.getApplicationPolicy()
				.getApplicationId());
	}

	@Override
	public void onElectionEvent(EventType eventType) {
		// TODO Auto-generated method stub
		logger.info("Event type: " + eventType.toString());
		if (eventType.toString().equals("ELECTED_COMPLETE")
				&& !messageBroker.isLeader()) {
			messageBroker.setLeader(true);
			messageBroker.setLeaderBroker(messageBroker.getMyHostName());
			logger.info("I am the new leader!");

			try {
				logger.info("Writing my hostname to leader path"
						+ getLeaderPath());
				ZooKeeperUtil.createEphemeralPath(currentZooKeeper,
						getLeaderPath(), messageBroker.getMyHostName());
			} catch (KeeperException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			messageBroker.getHeartBeatThread().run();

			// Do other stuffs since this is a new leader
		} else {

			try {
				if (ZooKeeperUtil.doesExistPath(currentZooKeeper, getLeaderPath())) {
					String leaderBroker = ZooKeeperUtil.readData(
							currentZooKeeper, getLeaderPath());
					messageBroker.setLeaderBroker(leaderBroker);
				}
			} catch (KeeperException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			} catch (InterruptedException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}

			// Why would this happen
			messageBroker.setLeader(false);
		}
	}

	public String getLeaderPath() {
		return leaderPath;
	}

	public void setLeaderPath(String leaderPath) {
		this.leaderPath = leaderPath;
	}

}
